package com.atguigu.upp.app;

import com.atguigu.upp.bean.TagInfo;
import com.atguigu.upp.service.MysqlDBService;
import com.atguigu.upp.utils.PropertiesUtil;
import com.atguigu.upp.utils.UPPUtil;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;

/**
 * Created by Smexy on 2022/11/12
 *
 select
 *
 from (
 select  uid,tagValue, 'tag_consuming_behavior_order_amount7d' tag from tag_consuming_behavior_order_amount7d where dt='2020-06-14'
 union all
 select  uid,tagValue, 'tag_population_attributes_nature_gender' tag from tag_population_attributes_nature_gender where dt='2020-06-14'
 union all
 select  uid,tagValue ,'tag_population_attributes_nature_period' tag from tag_population_attributes_nature_period where dt='2020-06-14'

 )tmp
 pivot(
 min(tagValue)
 for tag
 in ('tag_consuming_behavior_order_amount7d','tag_population_attributes_nature_gender','tag_population_attributes_nature_period')

 )
 */
public class MergeWideTableApp
{
    public static void main(String[] args) throws IOException {

        //接收参数
        String taskId = args[0];
        String doDate = args[1];

        //准备编程环境
        SqlSessionFactory sqlSessionFactory = UPPUtil.getSqlSessionFactoryByConfig("config.xml");
        MysqlDBService mysqlDBService = new MysqlDBService(sqlSessionFactory.openSession());
        SparkSession sparkSession = UPPUtil.getSparkSession("MergeWideTableApp");

        //查询今天要合并的标签
        List<TagInfo> tags = mysqlDBService.getTagInfoTodayExecute();

        //生成宽表的建表语句
        String createTableSql = generateCreateTableSql(doDate, tags);
        String insertSql = generateInsertSql(doDate, tags);

        //生成宽表的导数语句
        sparkSession.sql(createTableSql);
        sparkSession.sql(insertSql);


    }

    /*

    create table if not exists upp220704.widetable_2020_06_14(
    uid string,
    tag_consuming_behavior_order_amount7d string,
    tag_population_attributes_nature_gender string,
    tag_population_attributes_nature_period string
)
comment '2020-06-14标签宽表'
row format delimited fields terminated by '\t'
location '';
     */
    public static String generateCreateTableSql(String doDate,List<TagInfo> tags){
        String template =  " create table if not exists %s.%s(uid string, %s )    " +
                           "comment '标签宽表' " +
                            "row format delimited fields terminated by '\\t' " +
                            "location '%s/%s' ";

        //获取库名
        String dbName = PropertiesUtil.getProperty("updbname");
        String tableName = PropertiesUtil.getProperty("upwideprefix") + doDate.replace("-","_");
        String hdfsPath = PropertiesUtil.getProperty("hdfsPath");

        //生成标签列的信息
        String columnStr = tags.stream()
                             .map(t -> t.getTagCode().toUpperCase().concat(" string"))
                             .collect(Collectors.joining(","));

        String sql = String.format(template, dbName, tableName, columnStr, hdfsPath + dbName, tableName);

        System.out.println(sql);

        return sql;

    }


    /*
    insert overwrite table upp220704.up_tag_merge_2020_06_14
select
    *
from (
        select  uid,tagValue, 'tag_consuming_behavior_order_amount7d' tag from tag_consuming_behavior_order_amount7d where dt='2020-06-14'
        union all
        select  uid,tagValue, 'tag_population_attributes_nature_gender' tag from tag_population_attributes_nature_gender where dt='2020-06-14'
        union all
        select  uid,tagValue ,'tag_population_attributes_nature_period' tag from tag_population_attributes_nature_period where dt='2020-06-14'

    )tmp
pivot(
    min(tagValue)
    for tag
    in ('tag_consuming_behavior_order_amount7d','tag_population_attributes_nature_gender','tag_population_attributes_nature_period')

)

     */
    public static String generateInsertSql(String doDate,List<TagInfo> tags){

        String template = " insert overwrite table %s.%s  " +
                          " select * from ( %s ) tmp pivot ( min(tagValue)  for tag in ( %s )    )  ";


        //获取库名
        String dbName = PropertiesUtil.getProperty("updbname");
        String tableName = PropertiesUtil.getProperty("upwideprefix") + doDate.replace("-","_");

        //生成in部分，旋转列值
        String inStr = tags.stream()
                               .map(t -> "'" + t.getTagCode().toLowerCase()+ "'")
                               .collect(Collectors.joining(","));

        //tmp的单表
        String s = " select  uid,tagValue, '%s' tag from %s.%s where dt='%s'";
        //生成tmp
        String tmpStr = tags.stream()
                           .map(t -> String.format(s,t.getTagCode().toLowerCase(),dbName,t.getTagCode().toLowerCase(),doDate))
                           .collect(Collectors.joining(" union all "));

        //生成最终的sql
        String sql = String.format(template, dbName, tableName, tmpStr, inStr);

        System.out.println(sql);

        return sql;


    }
}
